package io.pumelo.noti.client;

import com.alibaba.fastjson.JSON;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.pumelo.noti.bean.EventsEnum;
import io.pumelo.noti.dto.BaseMsg;
import io.pumelo.noti.dto.LoginDto;
import io.pumelo.noti.properties.NotiServiceProperties;
import io.pumelo.noti.common.LOG;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

@Service
public class NettyClient {
    @Autowired
    private NotiServiceProperties properties;
    @Autowired
    private NotiChannelInitializer notiChannelInitializer;
    @Autowired
    private HeartBeat heartBeat;
    private Channel channel;
    private EventLoopGroup group;
    private long lastHeartBeatTime;
    private static final Integer threadCount = 20;
    private boolean lock = false;
    //heartbeat
    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    private ScheduledFuture<?> scheduledFuture;

    public synchronized void initConnect() throws InterruptedException {
        group = new NioEventLoopGroup(threadCount);
        final Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(notiChannelInitializer);
        LOG.info(this,"host:"+properties.getHost() + " port:"+properties.getPort());
        ChannelFuture future = bootstrap.connect(properties.getHost(), properties.getPort()).sync();
        if(future.isSuccess()){
            LOG.info(this,"connect success");
            channel = future.channel();
        }
    }

    /**
     * 初始化登录JSON
     * 发送给服务器
     */
    public synchronized void login(){
        LoginDto loginDtoElecProduct = new LoginDto(properties.getProductKey(),
                properties.getAuthId(),
                properties.getAuthSecret(),
                properties.getSubkey(),
                Arrays.asList(EventsEnum.STATUS_KV.getName(),EventsEnum.STATUS_RAW.getName(), EventsEnum.ONLINE.getName(), EventsEnum.OFFLINE.getName()));
        BaseMsg<LoginDto> loginMsg = new BaseMsg<>("login_req", Arrays.asList(loginDtoElecProduct));
        String msgStr = JSON.toJSONString(loginMsg)+"\n";
        LOG.warn(this,"send login msg:"+msgStr);
        sendMsg(msgStr);
    }

    public void sendMsg(String msg){
        channel.writeAndFlush(msg);
    }

    /**
     * 支持自动重连
     * 重新尝试间隔为5秒
     */
    public synchronized void init(){
        if(lock){
            return;
        }
        lock=true;
        try{
            this.initConnect();//初始化服务器信息
            this.login();//登录服务器
        }catch (Exception e){
            LOG.error(this,"try again 5s latter",e);
            try {
                Thread.sleep(5000);
            } catch (InterruptedException ignored) {}
            //开始循环重连
            innerInit();
        }
    }

    /**
     * 内部重连循环
     */
    private synchronized void innerInit(){
        try{
            this.initConnect();//初始化服务器信息
            this.login();//登录服务器
        }catch (Exception e){
            LOG.error(this,"try again 5s latter",e);
            try {
                Thread.sleep(5000);
            } catch (InterruptedException ignored) {}
            innerInit();
        }
    }

    public synchronized void destroy(){
        if(lock){
            return;
        }
        stopHeartBeat();
        if(channel.isOpen()){
            channel.close();
            LOG.info(this,"close channel");
        }
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException ignored) {}
        if(!group.isShutdown()){
            group.shutdownGracefully();
            LOG.info(this,"shutdown group gracefully,maybe not immediately");
        }

    }

    public long getLastHeartBeatTime() {
        return lastHeartBeatTime;
    }

    public void setLastHeartBeatTime(long lastHeartBeatTime) {
        this.lastHeartBeatTime = lastHeartBeatTime;
    }

    public boolean isLock() {
        return lock;
    }

    public void unLock() {
        this.lock = false;
    }

    public void startHeartBeat(ChannelHandlerContext ctx){
        heartBeat.initChannelHandlerContext(ctx);
        this.scheduledFuture = scheduledExecutorService.scheduleWithFixedDelay(heartBeat, 60, 60, TimeUnit.SECONDS);
        setLastHeartBeatTime(System.currentTimeMillis());
        LOG.info(this,"start heartbeat");
    }

    public void stopHeartBeat(){
        if(this.scheduledFuture != null){
            this.scheduledFuture.cancel(true);
            this.scheduledFuture = null;
            LOG.info(this,"stop heartbeat");
        }
    }
}

